package com.wei.web.middleware.retry.compensate;

import com.alibaba.com.caucho.hessian.io.Hessian2Input;
import com.alibaba.com.caucho.hessian.io.Hessian2Output;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.tuya.biz.platform.framework.compensate.aspect.RetryContext;
import com.tuya.biz.platform.framework.compensate.config.CompensateConfig;
import com.tuya.biz.platform.framework.compensate.dataobject.MqCompensationDO;
import com.tuya.biz.platform.framework.compensate.distribution.NodeSlice;
import com.tuya.biz.platform.framework.compensate.distribution.NodeSliceModel;
import com.tuya.biz.platform.framework.compensate.mapper.MqCompensationMapper;
import com.tuya.biz.platform.framework.compensate.reach.Reach;
import com.tuya.biz.platform.framework.config.EnvironmentContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.core.GenericTypeResolver;
import org.springframework.scheduling.annotation.Scheduled;

import javax.annotation.PostConstruct;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;
import java.time.Instant;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author shihuimiao
 * create time: 上午9:54
 */
@Slf4j
public class CompensateService {
    private final int timesRequired;

    private final Integer waitSecond;

    private final Boolean open;

    private final MqCompensationMapper mqCompensationMapper;

    private final Reach reach;

    private final NodeSlice nodeSlice;

    private final Boolean dataIsolation;

    public final static String CLASS_METHOD_SEP = "&cms&";

    public CompensateService(CompensateConfig config,
                             MqCompensationMapper mqCompensationMapper,
                             Reach reach,
                             NodeSlice nodeSlice) {
        this.timesRequired = config.getRetry();
        if (config.getCycleSecond() == null) {
            this.waitSecond = config.getCycleMinute() * 60;
        } else {
            this.waitSecond = config.getCycleSecond();
        }
        this.open = config.getOpen();
        this.mqCompensationMapper = mqCompensationMapper;
        this.reach = reach;
        this.nodeSlice = nodeSlice;
        this.dataIsolation = config.getDataIsolation();
    }

    @PostConstruct
    public void doWork() {
        if (!open) {
            log.info("compensate work is close.");
            return;
        }

        ExecutorService asyncThreadCompensate = Executors.newSingleThreadExecutor(new BasicThreadFactory.Builder().namingPattern("async thread compensate worker").build());
        CompletableFuture.runAsync(() -> {
            Long start = 0L;
            while (true) {
                try {
                    List<MqCompensationDO> mqCompensationDOS = mqCompensationMapper.selectFailedWorkGtId(start, dataIsolation);
                    if (CollectionUtils.isEmpty(mqCompensationDOS)) {
                        start = 0L;
                        TimeUnit.SECONDS.sleep(this.waitSecond);
                    } else {
                        NodeSliceModel sliceModel = nodeSlice.getSliceModel();
                        log.info("compensate info slice mode ={}", sliceModel);
                        for (MqCompensationDO mqCompensationDO : mqCompensationDOS) {
                            if (hit(mqCompensationDO.getId(), sliceModel)) {
                                if (mqCompensationDO.getCallClass().contains(CLASS_METHOD_SEP)) {
                                    retryWorkWithAnnotation(mqCompensationDO);
                                } else {
                                    retryWork(mqCompensationDO);
                                }
                            }
                        }
                        MqCompensationDO mqCompensationDO = mqCompensationDOS.get(mqCompensationDOS.size() - 1);
                        start = mqCompensationDO.getId();
                    }
                } catch (Exception e) {
                    log.error("do work loop error", e);
                    try {
                        TimeUnit.SECONDS.sleep(this.waitSecond);
                    } catch (InterruptedException ex) {
                    }
                }
            }
        }, asyncThreadCompensate);
    }

    private boolean hit(Long id, NodeSliceModel sliceModel) {
        long mod = id % sliceModel.getTotal();
        return sliceModel.getNum().equals(Integer.parseInt(String.valueOf(mod)));
    }

    public void saveTask(String key, Object param, String clazz, Integer timesRequired) throws Exception {
        MqCompensationDO mqCompensationDO = new MqCompensationDO();
        mqCompensationDO.setUniqueKey(key);
        mqCompensationDO.setTimesRequired(Objects.isNull(timesRequired) ? this.timesRequired : timesRequired);
        mqCompensationDO.setRetryTime(0);
        mqCompensationDO.setParam(JSON.toJSONString(param, SerializerFeature.WriteClassName));
        mqCompensationDO.setCallClass(clazz);
        mqCompensationDO.setSuccess(false);
        mqCompensationDO.setSuspended(false);
        mqCompensationDO.setIsPre(EnvironmentContext.isPre());
        this.mqCompensationMapper.insert(mqCompensationDO);
    }


    private String encode(Object param) throws IOException {
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        Hessian2Output output = new Hessian2Output(os);
        output.writeObject(param);
        output.close();
        return new String(os.toByteArray(), "ISO-8859-1");
    }

    private Object decode(String param) throws IOException {
        byte[] bytes = param.getBytes("ISO-8859-1");
        ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
        Hessian2Input input = new Hessian2Input(bis);
        return input.readObject();
    }


    public void retryWork(MqCompensationDO mqCompensationDO) {
        log.debug("work mqCompensationDO={}", mqCompensationDO);
        if (!updateRetryTimeByCAS(mqCompensationDO)) {
            return;
        }
        String clazz = mqCompensationDO.getCallClass();
        String param = mqCompensationDO.getParam();
        boolean isSuccess = false;
        try {
            Class<?> beanClass = Class.forName(clazz);
            Object bean = EnvironmentContext.getBean(beanClass);
            if (!(Processor.class.isAssignableFrom(beanClass))) {
                log.error("work error bean not exist! mqCompensationDO={},beanClass={}", mqCompensationDO, beanClass);
                mqCompensationMapper.suspendByPrimaryKey(mqCompensationDO.getId());
                return;
            }
            Processor actuator = (Processor) bean;
            Class actualType = getActualType(actuator.getClass());
            Object paramBean = JSON.parseObject(param, actualType);
            isSuccess = actuator.execute(paramBean);
        } catch (Exception e) {
            log.error("work unexpect error ! mqCompensationDO={}", mqCompensationDO, e);
        } finally {
            mqCompensationMapper.updateResultById(isSuccess, mqCompensationDO.getId());
            if (!isSuccess && mqCompensationDO.getTimesRequired() <= mqCompensationDO.getRetryTime() + 1) {
                log.warn("times required is over but still failed;mqCompensationDO={}", mqCompensationDO);
                singleReach(mqCompensationDO.getUniqueKey());
            }
        }
    }

    public void retryWorkWithAnnotation(MqCompensationDO mqCompensationDO) {
        log.debug("work mqCompensationDO={}", mqCompensationDO);
        if (!updateRetryTimeByCAS(mqCompensationDO)) {
            return;
        }
        String clazz = mqCompensationDO.getCallClass();
        String param = mqCompensationDO.getParam();

        String[] split = clazz.split(CLASS_METHOD_SEP);
        if (split.length != 2) {
            log.error("split class method error; clazz={}", clazz);
        }

        String c = split[0];
        String m = split[1];

        boolean isSuccess = false;
        try {
            Class<?> beanClass = Class.forName(c);
            Object bean = EnvironmentContext.getBean(beanClass);
            Object[] objArrType = {};
            Object[] objects = JSON.parseObject(param, objArrType.getClass());
            Class<?>[] cArr = new Class[objects.length];
            for (int i = 0; i < objects.length; i++) {
                cArr[i] = objects[i].getClass();
            }
            Method declaredMethod = beanClass.getDeclaredMethod(m, cArr);
            RetryContext.setNotFirst();
            declaredMethod.invoke(bean, objects);
            isSuccess = true;
        } catch (Exception e) {
            log.error("work unexpect error ! mqCompensationDO={}", mqCompensationDO, e);
        } finally {
            RetryContext.clear();
            mqCompensationMapper.updateResultById(isSuccess, mqCompensationDO.getId());
            if (!isSuccess && mqCompensationDO.getTimesRequired() <= mqCompensationDO.getRetryTime() + 1) {
                log.warn("times required is over but still failed;mqCompensationDO={}", mqCompensationDO);
                singleReach(mqCompensationDO.getUniqueKey());
            }
        }
    }

    private Class getActualType(Class clazz) {
        return GenericTypeResolver.resolveTypeArgument(clazz, Processor.class);
    }

    /**
     * 不是用加锁方案，使用cas处理
     * 因为是分布式系统 这条可能被其他节点更新处理过了
     *
     * @param mqCompensationDO
     * @return
     */
    private boolean updateRetryTimeByCAS(MqCompensationDO mqCompensationDO) {
        return mqCompensationMapper.updateRetryTimeByCAS(mqCompensationDO.getRetryTime(), mqCompensationDO.getId()) > 0;
    }

    private void singleReach(String uniqueKey) {
        String content = "【消息补偿-补偿失败】，请相关同事注意。\n" +
                ">应用 : <font color=\"comment\">" + EnvironmentContext.getAppName() + "</font> \n" +
                ">环境 : <font color=\"comment\">" + EnvironmentContext.getEnv() + "</font> \n" +
                ">区域 : <font color=\"comment\">" + EnvironmentContext.getDeployRegion() + "</font> \n" +
                ">消息key : <font color=\"comment\">" + uniqueKey + "</font> \n";

        this.reach.send(content);
    }

    /**
     * 统计通知
     */
    @Scheduled(cron = "0 0 10 * * ?")
    private void multiReach() {
        if (!open || !nodeSlice.getSliceModel().getNum().equals(0)) return;

        long nowTime = Instant.now().toEpochMilli();
        long lastDayTime = nowTime - 24 * 60 * 60 * 1000;

        int count = mqCompensationMapper.failedCountByTime(lastDayTime, dataIsolation);
        if (count == 0) return;

        String content = "【消息补偿-失败统计】，请相关同事注意。\n" +
                ">时间范围 : <font color=\"comment\">24小时内</font> \n" +
                ">应用 : <font color=\"comment\">" + EnvironmentContext.getAppName() + "</font> \n" +
                ">环境 : <font color=\"comment\">" + EnvironmentContext.getEnv() + "</font> \n" +
                ">区域 : <font color=\"comment\">" + EnvironmentContext.getDeployRegion() + "</font> \n" +
                ">失败条数 : <font color=\"comment\">" + count + "</font> \n";

        this.reach.send(content);
    }

}
